#include "service_discovery.h"

// time function' header.
#if defined(_WIN32)
 #include <windows.h>
 #include "time.h"
 #include <sys/timeb.h>
 #include <direct.h>
 #include <io.h>
 #pragma warning(disable:4996)
#else
 #include <unistd.h>
 #include <sys/stat.h>
 #include <time.h>
 #include <arpa/inet.h>
 #include <unistd.h>
 #include <sys/time.h>
#endif

namespace anet {
	namespace ServiceDiscoverySpace {
		// sleep ms.
        #define sleep_ms(x) std::this_thread::sleep_for(std::chrono::milliseconds(x))

		std::atomic_bool gOpen = true;

		// get time info: {second, millisecond}
		inline std::pair<time_t, int> getTimeInfo() {
        #if defined(_WIN32) || defined(_WIN64)
			struct timeb tp;
			ftime(&tp);
			return { tp.time,tp.millitm };
        #else
			struct timeval now;
			struct timezone zone;
			gettimeofday(&now, &zone);
			return { now.tv_sec,now.tv_usec / 1000 };
        #endif
		}

		// get current time.
		template <size_t N>
		inline const char* buildCurrentTime(char(&timeInfo)[N]) {
			auto timePair = getTimeInfo();
			auto ms = timePair.second;
			auto s = timePair.first;
			auto tm = localtime(&s);
			int n = std::snprintf(timeInfo, sizeof(timeInfo),
				"%d-%02d-%02d %02d:%02d:%02d.%03d",
				1900 + tm->tm_year, tm->tm_mon + 1, tm->tm_mday,
				tm->tm_hour, tm->tm_min, tm->tm_sec, ms
			);
			assert(n > 0 && n <= int(sizeof(timeInfo)));
			return timeInfo;
		}

		// output to stdout.
        #define printArgs(format) {         \
		    va_list args;                   \
		    va_start(args, format);         \
		    vfprintf(stdout, format, args); \
		    va_end(args);                   \
	    }

		// default implementation of ILog for internal log.
		class defaultServiceDiscoveryLog : public IServiceDiscoveryLog {
		public:
			defaultServiceDiscoveryLog() = default;
			virtual ~defaultServiceDiscoveryLog() = default;

		public:
			virtual bool setLevel(int level) {
				return true;
			}
			virtual int getLevel() const {
				return 0;
			}
			virtual void Debug(const char* format, ...) override {
				if (!gOpen) {
					return;
				}

				char time[1024];
				buildCurrentTime(time);
				fprintf(stdout, "%s ", time);

				printArgs(format);
				fprintf(stdout, "%s", "\n");
			}
			virtual void Info(const char* format, ...) override {
				if (!gOpen) {
					return;
				}

				char time[1024];
				buildCurrentTime(time);
				fprintf(stdout, "%s ", time);

				printArgs(format);
				fprintf(stdout, "%s", "\n");
			}
			virtual void Warn(const char* format, ...) override {
				if (!gOpen) {
					return;
				}

				char time[1024];
				buildCurrentTime(time);
				fprintf(stdout, "%s ", time);

				printArgs(format);
				fprintf(stdout, "%s", "\n");
			}
			virtual void Crit(const char* format, ...) override {
				if (!gOpen) {
					return;
				}

				char time[1024];
				buildCurrentTime(time);
				fprintf(stdout, "%s ", time);

				printArgs(format);
				fprintf(stdout, "%s", "\n");
			}
		};

		// declare a global default log.
		static defaultServiceDiscoveryLog internalLog;
		IServiceDiscoveryLog* gLog = &internalLog;

		// ServiceDiscovery implementation.
		ServiceDiscovery::ServiceDiscovery(const std::string& name,
			const std::string& endpoint, const std::string& attach,
			newServiceDiscoveryFunc newServiceDiscovery) :
			m_newServiceDiscoveryFunc(std::move(newServiceDiscovery)),
			m_subscribeCount(0), m_wheel(gTimerWheelSize),
			m_timer(m_wheel) {

			// initialize my node information.
			m_selfEntity.name = name;
			m_selfEntity.endpoint = endpoint;
			m_selfEntity.attach = attach;
		}

		ServiceDiscovery::ServiceDiscovery(const ServiceEntityInfo& entityInfo, 
			newServiceDiscoveryFunc newServiceDiscovery) : 
			m_newServiceDiscoveryFunc(std::move(newServiceDiscovery)), 
			m_subscribeCount(0), m_wheel(gTimerWheelSize),
			m_timer(m_wheel) {
			m_selfEntity = entityInfo;
		}

		ServiceDiscovery::~ServiceDiscovery() {
			this->Release();
		}

		bool ServiceDiscovery::Init(std::string serverIP, uint16 serverPort, std::string passwd, int index,
			uint32 publishInterval, const std::string& redisKey
		) {
			// redis key.
			m_serviceDiscoveryKey = redisKey;

			// try to connect the redis(ip:port).
			try {
				if (!m_redisClient.Init(serverIP, serverPort, passwd, index, 0, 4)) {
					return false;
				}
			} catch (...) {
				return false;
			}

			// call redis interval
			if (publishInterval < gSubscribeTimeInterval) {
				publishInterval = gSubscribeTimeInterval;
			}

			// log out.
			gLog->Debug("connect %s:%d redis server successfully.", serverIP.c_str(), serverPort);

			// create io_context and its worker.
			m_ioContext = std::make_unique<asio::io_context>();
			m_worker = std::make_unique<asio::io_context::work>(*m_ioContext);

			// start a thread to run the redis operation.
			m_th = std::make_unique<std::thread>([this]() {
				this->m_ioContext->run();
			});

			// Start timer to subscribe and publish.
			m_timer.add_repeated_timer([this](void* para) {
				this->timerDo(); // timer callback function.
			}, gSubscribeTimerId, publishInterval);

			return true;
		}

		// update timer, try to sleep as possible.
		void ServiceDiscovery::Run(int count) {
			return m_wheel.update(count);
		}

		// Release releases all my resources.
		void ServiceDiscovery::Release() {
			if (m_ioContext != nullptr) {
				auto ret = m_redisClient.HDel(m_serviceDiscoveryKey, m_selfEntity.name);
				if (!ret) {
					gLog->Crit("remove %s:%s fail", m_serviceDiscoveryKey, m_selfEntity.name);
				}
				// stop IO context.
				m_ioContext->stop();
			}

			// thread to exit.
			if (m_th != nullptr) {
				m_th->join();
			}
		}

		// RemoveService removes service name from my node list.
		bool ServiceDiscovery::RemoveService(const std::string& name) {
			std::lock_guard<std::mutex> guard(m_mutex);
			return this->removeService(name);
		}

		// FindService is whether there is name node in my node list.
		bool ServiceDiscovery::FindService(const std::string& name) const {
			std::lock_guard<std::mutex> guard(m_mutex);
			for (size_t i = 0; i < m_serviceEntities.size(); i++) {
				if (m_serviceEntities[i].name == name) {
					return true;
				}
			}
			return false;
		}

		// subscribe subscribes from redis to get new service node in m_th thread.
		void ServiceDiscovery::subscribe() {
			enum findResult { // find result enum
				hasNoResult = -1,
				hasResult = 0,
				hasButDifferentResult = 1,
			};

			// try to get key's list and call user's callback when find new node.
			auto res = m_redisClient.HGetAll(m_serviceDiscoveryKey);
			if (res.first && res.second.type == redisSpace::RedisResultType_StringMap) {
				auto& map = res.second.mapResult;
				for (auto it = map.begin(); it != map.end(); it++) {
					const auto& key = it->first;
					const auto& value = it->second;
					if (key != m_selfEntity.name) {
						int findRet = hasNoResult;
						size_t index = 0;
						for (size_t j = 0; j < m_serviceEntities.size(); j++) {
							if (m_serviceEntities[j].name == key &&
								m_serviceEntities[j].endpoint == value) {
								findRet = hasResult;
								break;
							} else if (m_serviceEntities[j].name == key) {
								findRet = hasButDifferentResult;
								index = j;
								break;
							}
						}

						// result check.
						if (findRet == hasResult) {
							continue;
						} else {
							ServiceEntityInfo newNode{ key,value };
							if (findRet == hasNoResult) {
								m_serviceEntities.push_back(newNode);
							} else if (findRet == hasButDifferentResult) {
								m_serviceEntities[index].endpoint = value;
							}

							// Call user service discovery interface
							if (m_newServiceDiscoveryFunc != nullptr) {
								if (!m_newServiceDiscoveryFunc(&newNode)) {
									auto ret = this->removeService(newNode.name);
									if (!ret) {
										gLog->Crit("remove %s error", newNode.name.c_str());
									}
								}
							}
						}
					}
				}
			} else {
				gLog->Crit("redis hgetall error");
			}
		}

		// publish publishes myself entity information to redis in m_th thread.
		bool ServiceDiscovery::publish() {
			return m_redisClient.HSet(m_serviceDiscoveryKey, m_selfEntity.name, m_selfEntity.endpoint);
		}

		// timer's callback function in main thread.
		void ServiceDiscovery::timerDo() {
			// try to subscribe.
			asio::post(*m_ioContext, [this] {
				this->subscribe();
			});

			// try to publish my node information.
			if (m_subscribeCount <= gPublishMaxCount) {
				m_subscribeCount++;
				asio::post(*m_ioContext, [this] {
					this->publish();
				});
			}
		}

		// internal remove service which needs no lock.
		bool ServiceDiscovery::removeService(const std::string& name) {
			bool find = false;
			for (size_t i = 0; i < m_serviceEntities.size(); i++) {
				if (m_serviceEntities[i].name == name) {
					// remove the i element.
					auto allSize = m_serviceEntities.size() - 1;
					m_serviceEntities[i] = m_serviceEntities[allSize];
					m_serviceEntities.pop_back();
					find = true;
					break;
				}
			}
			return find;
		}
	}
}